package com.fanxuankai.boot.mqbroker.producer;

import cn.hutool.core.net.NetUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.fanxuankai.boot.mqbroker.Msg;
import com.fanxuankai.boot.mqbroker.producer.autoconfigure.MqBrokerProducerProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.temporal.ChronoUnit;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author fanxuankai
 */
public class MsgSendServiceImpl extends ServiceImpl<MsgSendMapper, MsgSend> implements MsgSendService {
    private static final Logger LOGGER = LoggerFactory.getLogger(MsgSendServiceImpl.class);

    @Resource
    private MqBrokerProducerProperties properties;
    @Resource
    private MqProducer<MsgSend> mqProducer;

    @Override
    public List<MsgSend> pullData() {
        return page(new Page<>(1, properties.getCompensation()
                .getBatchSize()), new QueryWrapper<MsgSend>().lambda().eq(Msg::getStatus, SendStatus.WAIT.getCode())
                .orderByAsc(Msg::getId).lt(Msg::getRetry, properties.getMaxRetry())).getRecords();
    }

    @Override
    public List<MsgSend> pullDelayedData() {
        return page(new Page<>(1, properties.getCompensation()
                .getBatchSize()), Wrappers.lambdaQuery(MsgSend.class).eq(Msg::getStatus, SendStatus.DELAYED.getCode())
                .le(MsgSend::getEffectTime, new Date())
                .lt(Msg::getRetry, properties.getMaxRetry())
                .orderByAsc(MsgSend::getEffectTime)).getRecords();
    }

    @Override
    public boolean lock(Long id, List<SendStatus> statusList) {
        MsgSend entity = new MsgSend();
        entity.setStatus(SendStatus.SENDING.getCode());
        entity.setLastModifiedDate(new Date());
        entity.setHostAddress(NetUtil.getLocalhostStr());
        if (CollectionUtils.isEmpty(statusList)) {
            statusList = Arrays.asList(SendStatus.DELAYED, SendStatus.WAIT);
        }
        return update(entity, Wrappers.lambdaUpdate(MsgSend.class).eq(Msg::getId, id)
                .in(Msg::getStatus, statusList.stream().map(SendStatus::getCode).collect(Collectors.toSet())));
    }

    @Override
    public void publisherCallbackTimeout() {
        Date timeout = Date.from(LocalDateTime.now().minus(properties.getCallbackTimeout(), ChronoUnit.MILLIS)
                .atZone(ZoneId.systemDefault()).toInstant());
        int lastChance = properties.getMaxRetry();
        while (true) {
            Page<MsgSend> page = page(new Page<>(1, 100), Wrappers.lambdaQuery(MsgSend.class)
                    .eq(Msg::getStatus, SendStatus.SENDING.getCode()).lt(Msg::getLastModifiedDate, timeout));
            if (page.getRecords().isEmpty()) {
                break;
            }
            MsgSend entity = new MsgSend();
            entity.setCause("回调超时");
            entity.setLastModifiedDate(new Date());
            for (MsgSend msgSend : page.getRecords()) {
                entity.setStatus(msgSend.getRetry() < lastChance ? SendStatus.WAIT.getCode() :
                        SendStatus.FAILURE.getCode());
                update(entity, Wrappers.lambdaQuery(MsgSend.class).eq(Msg::getId, msgSend.getId())
                        .eq(Msg::getStatus, SendStatus.SENDING.getCode())
                        .lt(msgSend.getRetry() < lastChance, Msg::getRetry, lastChance)
                        .ge(msgSend.getRetry() >= lastChance, Msg::getRetry, lastChance)
                        .lt(Msg::getLastModifiedDate, timeout));
            }
            if (!page.hasNext()) {
                break;
            }
        }
    }

    @Override
    public void success(MsgSend msg) {
        MsgSend entity = new MsgSend();
        entity.setLastModifiedDate(new Date());
        entity.setHostAddress(NetUtil.getLocalhostStr());
        entity.setStatus(SendStatus.SUCCESS.getCode());
        update(entity, Wrappers.lambdaUpdate(MsgSend.class).eq(Msg::getId, msg.getId())
                .eq(Msg::getStatus, SendStatus.SENDING.getCode()));
    }

    @Override
    public void success(String topic, String code) {
        MsgSend msgSend = getOne(Wrappers.lambdaQuery(MsgSend.class)
                .eq(Msg::getTopic, topic)
                .eq(Msg::getCode, code));
        if (msgSend != null) {
            success(msgSend);
        }
    }

    @Override
    public void failure(String topic, String code, String cause) {
        MsgSend msg = getOne(Wrappers.lambdaQuery(MsgSend.class).eq(Msg::getTopic, topic).eq(Msg::getCode, code));
        if (msg == null) {
            return;
        }
        msg.setCause(cause);
        failure(msg);
    }

    @Override
    public void failure(MsgSend msg) {
        MsgSend entity = new MsgSend();
        // 失败，尝试次数 +1
        entity.setRetry(msg.getRetry() + 1);
        entity.setCause(msg.getCause());
        entity.setLastModifiedDate(new Date());
        entity.setHostAddress(NetUtil.getLocalhostStr());
        LambdaUpdateWrapper<MsgSend> lambda = Wrappers.lambdaUpdate(MsgSend.class).eq(Msg::getId, msg.getId())
                .eq(Msg::getStatus, SendStatus.SENDING.getCode());
        int lastChance = properties.getMaxRetry();
        if (entity.getRetry() < lastChance) {
            entity.setStatus(SendStatus.WAIT.getCode());
        } else {
            entity.setStatus(SendStatus.FAILURE.getCode());
        }
        update(entity, lambda);
    }

    @Override
    public void compensate(List<Long> ids) {
        if (CollectionUtils.isEmpty(ids)) {
            return;
        }
        for (Long id : ids) {
            if (!lock(id, Collections.singletonList(SendStatus.SUCCESS))) {
                continue;
            }
            MsgSend msgSend = getById(id);
            mqProducer.produce(msgSend);
            LOGGER.info("消息补偿成功, topic: {}, code: {}", msgSend.getTopic(), msgSend.getCode());
        }
    }
}
